package com.loveprogrammer.base.network.support;

import com.alibaba.fastjson2.JSON;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.loveprogrammer.base.factory.HandlerFactory;
import com.loveprogrammer.base.factory.SpringContextHelper;
import com.loveprogrammer.command.anotation.TagListener;
import com.loveprogrammer.command.client.ClientLoginTag;
import com.loveprogrammer.command.client.ClientTopic;
import com.loveprogrammer.dto.base.RequestBase;
import com.loveprogrammer.listener.INetworkEventListener;
import com.loveprogrammer.constants.CommonValue;
import com.loveprogrammer.pojo.StringMessage;
import com.loveprogrammer.utils.ChannelUtils;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Component
public class NetworkListener implements INetworkEventListener {

    protected static final Logger logger = LoggerFactory.getLogger(NetworkListener.class);

    private final ThreadPoolExecutor executor = new ThreadPoolExecutor(
            1,
            10,
            0L,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(20480),
            new ThreadFactoryBuilder()
                    .setNameFormat("worker-pool-%d").build(),
            new ThreadPoolExecutor.CallerRunsPolicy()
    );

    @Autowired
    private HandlerFactory handlerFactory;


    @Override
    public void onConnected(ChannelHandlerContext ctx) {
        logger.info("建立连接");
        SessionManager.getInstance().create(ctx.channel());
    }

    @Override
    public void onDisconnected(ChannelHandlerContext ctx) {
        logger.info("建立断开");
        SessionManager.getInstance().close(ctx.channel());
    }

    @Override
    public void onExceptionCaught(ChannelHandlerContext ctx, Throwable throwable) {
        logger.warn("异常发生", throwable);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, StringMessage msg) {
        // 消息分发 对于消息的处理要增加多线程
        int topicId = msg.getTopicId();
        int tagId = msg.getTagId();

        Class handler = handlerFactory.handlerMap.get(topicId);
        if (handler == null) {
            logger.warn("未获取到指定的消息监听对象，topicId {}", topicId);
            return;
        }
        String bodyValue = msg.getBody();

        executor.execute(() -> {
            Object bean;
            Object param = null;
            try {
                bean = SpringContextHelper.getBean(handler);
                // 找到tag 遍历methods
                Method[] methods = handler.getMethods();
                StringMessage result = null;
                for (Method method : methods) {
                    TagListener mqListener = method.getAnnotation(TagListener.class);
                    if (tagId == mqListener.tag()) {
                        Class<?> aClass = mqListener.messageClass();
                        String name = aClass.getName();
                        // 先处理基本类型
                        if ("java.lang.String".equals(name)) {
                            result = (StringMessage) method.invoke(bean, ctx, bodyValue);
                        } else if ("java.lang.Long".equals(name)) {
                            Long object = Long.parseLong(bodyValue);
                            result = (StringMessage) method.invoke(bean, ctx, object);
                        } else if ("java.lang.Integer".equals(name)) {
                            Integer object = Integer.parseInt(bodyValue);
                            result = (StringMessage) method.invoke(bean, ctx, object);
                        } else if ("java.lang.Short".equals(name)) {
                            Short object = Short.parseShort(bodyValue);
                            result = (StringMessage) method.invoke(bean, ctx, object);
                        } else if ("java.lang.Byte".equals(name)) {
                            Byte object = Byte.parseByte(bodyValue);
                            result = (StringMessage) method.invoke(bean, ctx, object);
                        } else if ("java.lang.Double".equals(name)) {
                            Double object = Double.parseDouble(bodyValue);
                            result = (StringMessage) method.invoke(bean, ctx, object);
                        } else if ("java.lang.Float".equals(name)) {
                            Float object = Float.parseFloat(bodyValue);
                            result = (StringMessage) method.invoke(bean, ctx, object);
                        }
                        // 转对象类型
                        else {
                            param = JSON.parseObject(bodyValue, aClass);
                            result = (StringMessage) method.invoke(bean, ctx, param);
                        }
                        break;
                    }
                }
                if (result != null) {
                    ChannelUtils.pushToClient(ctx.channel(), result);
                } else {
                    logger.warn("未获取到指定的消息监听对象或者没有返回值，topicId {} tagId {}", topicId, tagId);
                }
            } catch (Exception e) {
                logger.error("发生异常", e);
                String result;
                if (e instanceof InvocationTargetException) {
                    InvocationTargetException ite = (InvocationTargetException) e;
                    Throwable exception = ite.getTargetException();
                    String message = exception.getMessage();
                    result = message;
                } else {
                    result = "异常:" + e.getMessage();
                }
                StringMessage message;
                RequestBase base = JSON.parseObject(bodyValue ,RequestBase.class);
                if (base != null && base.getErrorCallbackTopic() > 0 && base.getErrorCallbackTag() > 0) {
                    message = StringMessage.create(base.getErrorCallbackTopic(), base.getErrorCallbackTag());
                    base.setErrorMsg(result);
                    message.setBody(JSON.toJSONString(base));
                } else {
                    message = StringMessage.create(ClientTopic.TOPIC_LOGIN, ClientLoginTag.TAG_LOGIN_REGISTER);
                    message.setBody(result);
                }

                message.setStatusCode(CommonValue.MSG_STATUS_CODE_FAIL);
                ChannelUtils.pushToClient(ctx.channel(), message);
            }
        });

    }
}
